package com.corn.kafka.transacation;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;
import java.util.Map;

/**
 * @author : Jim Wu
 * @version 1.0
 * @function :
 * @since : 2022/6/27 18:09
 */

public class TransactionalMsg {

    public static void main(String[] args) {
        Map<String, Object> config = new HashMap<>();

        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092,node2:9092");
        config.put(ProducerConfig.ACKS_CONFIG, "all");
        config.put(ProducerConfig.RETRIES_CONFIG, 3);
        config.put(ProducerConfig.BATCH_SIZE_CONFIG, 163840);
        config.put(ProducerConfig.LINGER_MS_CONFIG, 10);
        config.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        config.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
        config.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "abc");

        KafkaProducer<String, String> producer = new KafkaProducer(config);

        // 初始化事务
        producer.initTransactions();

        // 开启事务
        producer.beginTransaction();
        String topic = "test_topic";

        try {
            for (int i = 0; i < 5; i++) {
                // 发送事务消息
                ProducerRecord<String, String> record = new ProducerRecord(topic, 1, "k1", "this is a transaction msg " + i);
                producer.send(record);
                if (i == 3) {
//                    int v = 1 / 0;
                }
            }  // 提交事务
            producer.commitTransaction();
        } catch (Exception e) {
            e.printStackTrace();
            producer.abortTransaction();
        } finally {
            producer.close();
        }
    }
}
